package com.atguigu.kafka.flink;

import com.atguigu.kafka.flink.model.Person;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.ZoneOffset;

public class ExtractMinMaxPersonFunction  extends KeyedProcessFunction<String, Person, Tuple2<String, Person>> {
    private ValueState<Person> earliestMessage;
    private ValueState<Person> latestMessage;
    private ValueState<Long> timerState; // 新增的状态，用于跟踪已注册的定时器时间


    @Override
    public void open(Configuration parameters) throws Exception {
        earliestMessage = getRuntimeContext().getState(new ValueStateDescriptor<>("earliest-message", TypeInformation.of(Person.class)));
        latestMessage = getRuntimeContext().getState(new ValueStateDescriptor<>("latest-message", TypeInformation.of(Person.class)));
        timerState = getRuntimeContext().getState(new ValueStateDescriptor<>("timer-state", Long.class));
    }
    @Override
    public void processElement(Person person, KeyedProcessFunction<String, Person, Tuple2<String, Person>>.Context ctx, Collector<Tuple2<String, Person>> out) throws Exception {
        // 打印当前水印
//        long currentWatermark = ctx.timerService().currentWatermark();
//        System.out.println("Current Watermark: " + currentWatermark);

        Person currentEarliest = earliestMessage.value();
        Person currentLatest = latestMessage.value();

        if (currentEarliest == null || person.getDatetime().isBefore(currentEarliest.getDatetime())) {
            earliestMessage.update(person);
        }

        if (currentLatest == null || person.getDatetime().isAfter(currentLatest.getDatetime())) {
            latestMessage.update(person);
        }

        //20秒之内最早的数据和最晚的数据
//        ctx.timerService().registerEventTimeTimer(person.getDatetime().plusDays(1).atZone(ZoneOffset.UTC).toInstant().toEpochMilli());
//        ctx.timerService().registerEventTimeTimer(person.getDatetime().plusSeconds(18).atZone(ZoneOffset.UTC).toInstant().toEpochMilli());
        // 获取事件时间
        long eventTime = person.getDatetime().atZone(ZoneOffset.systemDefault()).toInstant().toEpochMilli();
        Long windowEndTime = timerState.value();

        if (windowEndTime == null||eventTime>windowEndTime)  {
            // 如果还没有注册定时器，则注册一个
            ctx.timerService().registerEventTimeTimer(person.getDatetime().plusSeconds(10).atZone(ZoneOffset.systemDefault()).toInstant().toEpochMilli());
            timerState.update(person.getDatetime().plusSeconds(10).atZone(ZoneOffset.systemDefault()).toInstant().toEpochMilli());
            System.out.println("Registering new timer for time: " + person.getDatetime().plusSeconds(10).atZone(ZoneOffset.systemDefault()).toInstant().toEpochMilli());
        }

    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Person>> out) throws Exception {
        System.out.println("Timer triggered at time: " + timestamp);
        System.out.print("====>定时器：触发的watermark是："+ctx.timerService().currentWatermark());

        Person earliest = earliestMessage.value();
        Person latest = latestMessage.value();

        if (earliest != null) {
            out.collect(Tuple2.of(ctx.getCurrentKey(), earliest));
        }

        if (latest != null) {
            out.collect(Tuple2.of(ctx.getCurrentKey(), latest));
        }

        earliestMessage.clear();
        latestMessage.clear();
    }
}
